package chantask

import (
	"fmt"
	"gitee.com/ymofen/gobase"
	"gitee.com/ymofen/panicsafe"

	"sync"
	"sync/atomic"
	"time"
)

type chanTaskItem struct {
	fn   func(args ...interface{})
	args []interface{}
	t0   time.Time
}

type ChanTask struct {
	refcnt      int32
	pushcnt     int32
	popcnt      int32
	pushfailcnt int32
	id          string
	timeout     time.Duration
	wg          sync.WaitGroup
	terminated  int8
	taskChan    chan *chanTaskItem
}

var (
	checkQueueDuration = time.Millisecond * 200
)

func SetCheckQueueDuration(dura time.Duration) {
	checkQueueDuration = dura
}

func NewChanTask(maxCache int) *ChanTask {
	rval := &ChanTask{
		taskChan: make(chan *chanTaskItem, maxCache),
		timeout:  time.Second,
	}
	rval.start()
	return rval
}

func (this *ChanTask) IsTerminated() bool {
	return this.terminated == 1
}

func (this *ChanTask) innerExec(itm *chanTaskItem) {
	if panicsafe.GoFunCatchException {
		defer panicsafe.DeferCatchPanic()
	}
	if checkQueueDuration > 0 {
		dura := time.Since(itm.t0)
		if dura > checkQueueDuration {
			gobase.Warnf("The time from submission to execution is %dms.\tid=%s", dura.Milliseconds(), this.id)
		}
	}

	itm.fn(itm.args...)
}

func (this *ChanTask) innerWork() {
	if panicsafe.GoFunCatchException {
		defer panicsafe.DeferCatchPanic()
	}
	defer this.wg.Done()
break_for:
	for {
		select {
		case itm := <-this.taskChan:
			if itm == nil {
				break break_for
			}
			atomic.AddInt32(&this.popcnt, 1)
			this.innerExec(itm)
		}
	}
}

func (this *ChanTask) Remain() int32 {
	return atomic.LoadInt32(&this.pushcnt) - atomic.LoadInt32(&this.popcnt)
}

func (this *ChanTask) start() {
	this.wg.Add(1)
	go this.innerWork()
}

func (this *ChanTask) Stop() {
	this.terminated = 1

	select {
	case this.taskChan <- nil:
	case <-time.After(time.Second * 3):
		close(this.taskChan)
	}

	this.wg.Wait()
}

func (this *ChanTask) StatusString() string {
	cnt1, cnt2 := atomic.LoadInt32(&this.pushcnt), atomic.LoadInt32(&this.popcnt)
	return fmt.Sprintf("%d-%d=%d, fail:%d", cnt1, cnt2, cnt1-cnt2, this.pushfailcnt)
}

func (this *ChanTask) PushTask(fn func(args ...interface{}), args ...interface{}) error {
	itm := &chanTaskItem{fn: fn, args: args, t0: time.Now()}
	select {
	case this.taskChan <- itm:
		atomic.AddInt32(&this.pushcnt, 1)
	case <-time.After(this.timeout):
		atomic.AddInt32(&this.pushfailcnt, 1)
		return fmt.Errorf("ChanTask->PushTask timeout")
	}
	return nil
}
